package com.yihaodian.search.broker;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import com.yihaodian.api.search.SearchRequest;
import com.yihaodian.api.search.SearchResponse;
import com.yihaodian.api.search.facet.FacetResult;
import com.yihaodian.api.search.facet.SearchFacetSpec;
import com.yihaodian.api.search.grouping.GroupingResult;
import com.yihaodian.api.search.grouping.GroupingSpec;
import com.yihaodian.api.search.misc.SearchHit;
import com.yihaodian.search.broker.shard.ShardSearchResponse;
import com.yihaodian.search.facet.merger.Collector;
import com.yihaodian.search.facet.merger.MergeFactory;
import com.yihaodian.search.grouping.GroupingMerger;

/**
 * 
 */
public class ResultMerger {

	/**
	 * merge sort. Include SearchHit and Facet.
	 * 
	 * @param request
	 * @param shardResponse
	 * @return
	 */
	public static SearchResponse merge(SearchRequest request,
			List<ShardSearchResponse> shardResponse) {
		/**
		 * merge search hit
		 */
		SearchHit[] result = SearchHitMerger.merge(request, shardResponse);
		int totalHit = 0;
		for (ShardSearchResponse resp : shardResponse) {
			totalHit += resp.getSearchResponse().getTotalHit();
		}

		Map<String, FacetResult> facetMap = mergeFacet(request, shardResponse);
		GroupingResult[] groupResults = mergeGrouping(request, shardResponse);

		SearchResponse response = new SearchResponse(totalHit, result,
				facetMap, groupResults);
		String[] nodes = new String[shardResponse.size()];
		for (int i = 0; i < shardResponse.size(); i++) {
			SearchResponse shardResp = shardResponse.get(i).getSearchResponse();
			nodes[i] = shardResp.getNode().toString() + " "
					+ shardResponse.get(i).getShardId() + ", hit "
					+ shardResp.getTotalHit();
		}
		response.setShardNodes(new HashSet<String>(Arrays.asList(nodes)));
		return response;
	}

	static Map<String, FacetResult> mergeFacet(SearchRequest request,
			List<ShardSearchResponse> shardResponse) {
		Map<String, SearchFacetSpec> specMap = request.getFacetSpecMap();
		if (specMap == null)
			return null;

		// Merge and sort facets
		Map<String, FacetResult> resultMap = new HashMap<String, FacetResult>();
		// Collect the same facet to a list.
		Map<String, List<FacetResult>> facetsMap = Collector.collectFacets(
				request, shardResponse);
		Iterator<String> specIt = specMap.keySet().iterator();
		while (specIt.hasNext()) {
			String facetName = specIt.next();
			SearchFacetSpec spec = specMap.get(facetName);
			FacetResult facetResult = MergeFactory.getMerger(spec.getOrderBy())
					.merge(facetsMap.get(facetName), spec);
			resultMap.put(facetName, facetResult);
		}

		return resultMap;
	}

	static GroupingResult[] mergeGrouping(SearchRequest request,
			List<ShardSearchResponse> shardResponse) {
		GroupingSpec spec = request.getGroupingSpec();
		if (spec == null)
			return null;

		/**
		 * Merge grouping.
		 */
		List<GroupingResult[]> groupsList = new ArrayList<GroupingResult[]>();
		for (ShardSearchResponse resp : shardResponse) {
			GroupingResult[] groupResult = resp.getSearchResponse().getGroupingResult();
			if (groupResult != null && groupResult.length > 0) {
				groupsList.add(groupResult);
			}
		}
		GroupingResult[] groupResults = GroupingMerger.merge(groupsList);

		return groupResults;
	}
}
